#include "discovery/eventfd_domain_socket_sender.h"
#include "discovery/eventfd_types.h"
#include <cstring>
#include <cerrno>
#include <fcntl.h>
#include <iostream>
#include <sys/socket.h>
#include <sys/un.h>
#include <unistd.h>

#include "log/logger.h"
#include "pub_sub/pub_sub_infos.h"

EventFdDomainSocketSender::ConnectionInfo::ConnectionInfo()
    : socket_fd(-1), connect_time(std::chrono::steady_clock::now()) {}

void EventFdDomainSocketSender::ConnectionInfo::reset() {
  if (descriptor) {
    descriptor->close();
    descriptor.reset();
  }
  if (socket_fd >= 0) {
    close(socket_fd);
    socket_fd = -1;
  }
}

EventFdDomainSocketSender::EventFdDomainSocketSender(
    io_context &io, const std::string &socket_path,
    const std::string &identifier)
    : EventFdSender(io, identifier), socket_path_(socket_path) {

  connect_timer_ = std::make_unique<steady_timer>(io_);
  retry_timer_ = std::make_unique<steady_timer>(io_);
}

EventFdDomainSocketSender::~EventFdDomainSocketSender() { stop(); }

bool EventFdDomainSocketSender::start() {
  if (running_.exchange(true)) {
    return false;
  }

  try {
    return start_client();
  } catch (const std::exception &e) {
    JC_ERROR("启动发送器失败: {}", e.what());
    running_ = false;
    return false;
  }
}

void EventFdDomainSocketSender::stop() {
  if (!running_.exchange(false)) {
    return;
  }

  if (connect_timer_)
    connect_timer_->cancel();
  if (retry_timer_)
    retry_timer_->cancel();

  std::lock_guard<std::mutex> lock(connections_mutex_);
  connections_.clear();
}

void EventFdDomainSocketSender::async_send_eventfd(int eventfd,
                                                   TransferCallback callback) {
  transfer_callback_ = std::move(callback);

  {
    std::unique_lock<std::mutex> lock(connections_mutex_);
    if (connections_.empty()) {
      lock.unlock();
      initiate_connection_with_timeout(eventfd);
      return;
    }
  }

  send_eventfd_to_connection(eventfd, 0);
}

void EventFdDomainSocketSender::async_send_eventfd_with_info(int eventfd,
                                                            const std::string& message_type,
                                                            const std::string& topic_name,
                                                            const std::string& custom_data,
                                                            uint32_t message_id,
                                                            uint32_t priority,
                                                            uint32_t flags,
                                                            TransferCallback callback) {
  if (callback) {
    transfer_callback_ = std::move(callback);
  }

  {
    std::unique_lock<std::mutex> lock(connections_mutex_);
    if (connections_.empty()) {
      lock.unlock();
      initiate_connection_with_timeout(eventfd);
      return;
    }
  }

  send_eventfd_to_connection_with_info(eventfd, 0, message_type, topic_name, custom_data, message_id, priority, flags);
}

size_t EventFdDomainSocketSender::connection_count() const {
  std::lock_guard<std::mutex> lock(connections_mutex_);
  return connections_.size();
}

bool EventFdDomainSocketSender::start_client() {
  JC_INFO("EventFD发送器启动，目标: {}", socket_path_);

  auto subs = SubInfosManager::getInstance().getSubscriberInfos();
  for (const auto &sub : subs) {
    if (sub.topic_name == "eventfd_transfer") {
      JC_INFO("EventFD发送器启动，目标: {}", sub.topic_name);
      async_send_eventfd(sub.eventfd, [topic_name = sub.topic_name](const boost::system::error_code &ec, int eventfd) {
        if (!ec) {
          JC_INFO("成功发送EventFD: {} 到 {}", eventfd, topic_name);
        } else {
          JC_ERROR("发送EventFD失败: {}", ec.message());
        }
      });
    }
  }
  return true;
}

void EventFdDomainSocketSender::initiate_connection_with_timeout(int eventfd) {

  connect_timer_->expires_after(connect_timeout_);
  connect_timer_->async_wait([this,
                              eventfd](const boost::system::error_code &ec) {
    if (!ec) {

      JC_ERROR("连接超时");
      if (transfer_callback_) {
        transfer_callback_(make_error_code(boost::system::errc::timed_out), -1);
      }
      schedule_retry(eventfd);
    }
  });

  create_client_connection(eventfd);
}

void EventFdDomainSocketSender::create_client_connection(int eventfd) {
  int client_socket = socket(AF_UNIX, SOCK_STREAM | SOCK_NONBLOCK, 0);
  if (client_socket < 0) {
    if (transfer_callback_) {
      transfer_callback_(
          make_error_code(boost::system::errc::no_such_file_or_directory), -1);
    }
    return;
  }

  struct sockaddr_un addr;
  memset(&addr, 0, sizeof(addr));
  addr.sun_family = AF_UNIX;
  strncpy(addr.sun_path, socket_path_.c_str(), sizeof(addr.sun_path) - 1);

  JC_DEBUG("连接到: {}", addr.sun_path);
  int result = connect(client_socket, (struct sockaddr *)&addr, sizeof(addr));
  if (result < 0 && errno != EINPROGRESS) {
    close(client_socket);
    schedule_retry(eventfd);
    return;
  }

  JC_DEBUG("连接成功: {}", addr.sun_path);
  auto connection = std::make_unique<ConnectionInfo>();
  connection->socket_fd = client_socket;
  connection->descriptor =
      std::make_unique<posix::stream_descriptor>(io_, client_socket);
  connection->peer_info = "server";

  connection->descriptor->async_wait(
      posix::stream_descriptor::wait_write,
      [this, connection_ptr = connection.get(),
       eventfd](const boost::system::error_code &ec) {
        connect_timer_->cancel();

        if (!ec) {

          JC_INFO("成功连接到服务端");

          if (connection_callback_) {
            connection_callback_(boost::system::error_code{}, true);
          }

          send_eventfd_to_connection(eventfd, 0);

          retry_count_ = 0;
        } else {
          JC_ERROR("连接失败: {}", ec.message());
          schedule_retry(eventfd);
        }
      });

  {
    std::lock_guard<std::mutex> lock(connections_mutex_);
    connections_.push_back(std::move(connection));
  }
}

void EventFdDomainSocketSender::send_eventfd_to_connection(
    int eventfd, size_t connection_index) {
  std::lock_guard<std::mutex> lock(connections_mutex_);

  if (connection_index >= connections_.size()) {
    if (transfer_callback_) {
      transfer_callback_(
          make_error_code(boost::system::errc::no_such_file_or_directory), -1);
    }
    return;
  }

  auto &connection = connections_[connection_index];

  TransferMessage msg;
  strncpy(msg.identifier, identifier_.c_str(), sizeof(msg.identifier) - 1);
  // 设置默认的消息信息
  msg.set_message_type("default");
  msg.set_topic_name("eventfd_transfer");
  msg.set_custom_data("basic_eventfd_message");
  msg.message_id = 0;
  msg.priority = 0;
  msg.flags = 0;

  if (send_fd_with_info(connection->socket_fd, eventfd, msg) > 0) {
    JC_INFO("成功发送EventFD: {} 到 {}", eventfd, connection->peer_info);

    if (transfer_callback_) {
      transfer_callback_(boost::system::error_code{}, eventfd);
    }
  } else {
    if (transfer_callback_) {
      transfer_callback_(make_error_code(boost::system::errc::broken_pipe), -1);
    }
  }
}

void EventFdDomainSocketSender::send_eventfd_to_connection_with_info(
    int eventfd, size_t connection_index,
    const std::string& message_type,
    const std::string& topic_name,
    const std::string& custom_data,
    uint32_t message_id,
    uint32_t priority,
    uint32_t flags) {
  std::lock_guard<std::mutex> lock(connections_mutex_);

  if (connection_index >= connections_.size()) {
    if (transfer_callback_) {
      transfer_callback_(
          make_error_code(boost::system::errc::no_such_file_or_directory), -1);
    }
    return;
  }

  auto &connection = connections_[connection_index];

  TransferMessage msg;
  strncpy(msg.identifier, identifier_.c_str(), sizeof(msg.identifier) - 1);
  msg.set_message_type(message_type);
  msg.set_topic_name(topic_name);
  msg.set_custom_data(custom_data);
  msg.message_id = message_id;
  msg.priority = priority;
  msg.flags = flags;

  if (send_fd_with_info(connection->socket_fd, eventfd, msg) > 0) {
    JC_INFO("成功发送EventFD: {} 到 {} (类型: {}, 主题: {}, 数据: {})", 
            eventfd, connection->peer_info, message_type, topic_name, custom_data);

    if (transfer_callback_) {
      transfer_callback_(boost::system::error_code{}, eventfd);
    }
  } else {
    if (transfer_callback_) {
      transfer_callback_(make_error_code(boost::system::errc::broken_pipe), -1);
    }
  }
}

void EventFdDomainSocketSender::schedule_retry(int eventfd) {
  if (retry_count_.load() >= max_retry_count_) {
    JC_ERROR("达到最大重试次数，放弃连接");
    if (transfer_callback_) {
      transfer_callback_(make_error_code(boost::system::errc::timed_out), -1);
    }
    return;
  }

  retry_count_++;
  retry_timer_->expires_after(retry_interval_);
  retry_timer_->async_wait(
      [this, eventfd](const boost::system::error_code &ec) {
        if (!ec && running_) {
          JC_INFO("重试连接 ({}/{})", retry_count_.load(), max_retry_count_);
          initiate_connection_with_timeout(eventfd);
        }
      });
}

int EventFdDomainSocketSender::send_fd(int socket_fd, int fd_to_send) {
  struct msghdr msg = {0};
  char buf[CMSG_SPACE(sizeof(fd_to_send))];
  memset(buf, '\0', sizeof(buf));

  struct iovec io = {.iov_base = (void *)"FD", .iov_len = 2};
  msg.msg_iov = &io;
  msg.msg_iovlen = 1;

  msg.msg_control = buf;
  msg.msg_controllen = sizeof(buf);

  struct cmsghdr *cmsg = CMSG_FIRSTHDR(&msg);
  cmsg->cmsg_level = SOL_SOCKET;
  cmsg->cmsg_type = SCM_RIGHTS;
  cmsg->cmsg_len = CMSG_LEN(sizeof(fd_to_send));

  *((int *)CMSG_DATA(cmsg)) = fd_to_send;

  return sendmsg(socket_fd, &msg, 0);
}

int EventFdDomainSocketSender::send_fd_with_info(int socket_fd, int fd_to_send, const TransferMessage& transfer_msg) {
  struct msghdr msg = {0};
  char buf[CMSG_SPACE(sizeof(fd_to_send))];
  memset(buf, '\0', sizeof(buf));

  // 使用TransferMessage作为消息数据
  struct iovec io = {.iov_base = (void *)&transfer_msg, .iov_len = sizeof(transfer_msg)};
  msg.msg_iov = &io;
  msg.msg_iovlen = 1;

  msg.msg_control = buf;
  msg.msg_controllen = sizeof(buf);

  struct cmsghdr *cmsg = CMSG_FIRSTHDR(&msg);
  cmsg->cmsg_level = SOL_SOCKET;
  cmsg->cmsg_type = SCM_RIGHTS;
  cmsg->cmsg_len = CMSG_LEN(sizeof(fd_to_send));

  *((int *)CMSG_DATA(cmsg)) = fd_to_send;

  return sendmsg(socket_fd, &msg, 0);
}
